package com.eshipenet.shipnet.dp_unpacker

import java.util.Properties

import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer

object EdgeChannelDataSinkProvider {

  def create(env: StreamExecutionEnvironment): SinkFunction[EdgeChannelData] = {
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "kafka-server:9092")
    properties.setProperty("group.id", "EdgeChannelDataProducer")
    val kafkaSink = new FlinkKafkaProducer[EdgeChannelData](
      "EdgeChannelData2",
      new EdgeChannelDataSerializer,
      properties,
      FlinkKafkaProducer.Semantic.EXACTLY_ONCE
    )
    kafkaSink
  }

}
